/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.commons.cli.BasicParser;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sss.client.Job;
import org.sss.client.JobEngine;
import org.sss.client.SssClient;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.Mapper;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.TupleGroupScaner;
import org.sss.mapreduce.datatype.PackableDouble;
import org.sss.mapreduce.datatype.PackableInt;

public class MatMul {
  static Logger logger = LoggerFactory.getLogger(MatMul.class);
  static final String SIZE = "SIZE";
  static final String INVEC = "INVEC";

  private static SimpleDateFormat dataFormatter = new SimpleDateFormat("yyyyMMddkkmmss");

  private static void printTimestamp(String what) {
    System.err.println("@ " + what  + " " + dataFormatter.format(new Date(System.currentTimeMillis())));
  }

  public static class MatMulMapper extends Mapper {
    double [] vector;

    @Override
    public void configure(Context context) throws Exception {

      Configuration config = context.getConfiguration();
      int size = Integer.parseInt((String)config.getObject(SIZE));
      vector = new double[size];
      logger.error("size = " + size);
      GroupID invecID = GroupID.createFromString((String) config.getObject(INVEC));
      logger.error("id = " + invecID.toString());

      // final AtomicInteger counter = new AtomicInteger(0);
      logger.error("sidedata start");
      context.scanSideData(invecID, PackableInt.class, PackableDouble.class,
          new TupleGroupScaner<PackableInt, PackableDouble>() {
            @Override
            public void set(PackableInt k , PackableDouble v) {
              // not guard
              vector[k.get()] = v.get();
              // counter.getAndIncrement();
            }
      });
      logger.error("sidedata done");

      /*
      Iterable<Tuple<PackableInt, PackableDouble>> iter =
          context.getTupleReader(invecID, PackableInt.class, PackableDouble.class);

      int counter = 0;
      for (Tuple<PackableInt, PackableDouble> kv: iter) {
        vector[kv.getKey().get()] = kv.getValue().get();
        counter ++;
        //logger.error("key = " + kv.getKey().get());
      }
      logger.error(counter + " keys read");
      */
      /*
      SideData<PackableInt, PackableDouble> sd =
          context.openSideData(invecID, PackableInt.class, PackableDouble.class);
      for (int i = 0; i < size; i++) {
        PackableInt k = new PackableInt(i);
        double v = sd.get(k).get(0).get();
        vector[k.get()] = v;
        logger.error("kv = " + k + "," + v);
      }
      */
    }

    public void map(Context context,
                    PackableInt key, PackableVector value,
                    Output<PackableInt, PackableDouble> output) throws Exception {
      double sum = 0.0;
      for (int i = 0; i < value.num; i++)
        sum += vector[value.indices[i]] * value.values[i];
      output.write(key, new PackableDouble(sum));
    }
  }

  public static void main(String[] args) throws Exception {
    String inputMat = "";
    String inVec = "";
    String size = "";
    boolean broadcast = false;
    final boolean destructive = false;

    SssClient client = new SssClient(args);

    String[] others = client.getArgs();
    BasicParser parser = new BasicParser();
    Options options = new Options();
    options.addOption("b", false, "broadcast sidedata.");
    try {
      CommandLine line = parser.parse(options, others);

      if (line.getArgs().length != 3) {
        System.err.println("usage: Matmul INPUTMAT_ID INPUTVEC_ID SIZE");
        System.exit(1);
      }
      broadcast = line.hasOption("b");

      inputMat = line.getArgs()[0];
      inVec = line.getArgs()[1];
      size = line.getArgs()[2];
      System.out.println("'" + inVec + "'");
    }
    catch(ParseException e) {
      System.err.println(e.getMessage());
      HelpFormatter f = new HelpFormatter();
      f.printHelp("Matmul", options);
      System.exit(1);
    }

    client.getConfiguration().set(SIZE, size);

    JobEngine engine = new JobEngine(client);
    GroupID cmat = GroupID.createFromString(inputMat);

    long before = System.currentTimeMillis();
    try {
      for (int i = 0; i < 10;++i) {
        printTimestamp("loop_start");
        GroupID coutv = GroupID.createRandom(engine, broadcast, destructive);

        Job.Builder jb = engine.getJobBuilder("jobA", MatMulMapper.class)
            .addInput("cmat", cmat).addOutput("coutv", coutv);
        jb.getConfiguration().set(INVEC, inVec);
        jb.build();

        engine.exec();
        long rm_before = System.currentTimeMillis();
        long nrm = client.removeTupleGroup(GroupID.createFromString(inVec));
        long rm_now = System.currentTimeMillis();
        System.out.println("output data deployed - " + coutv);

        long now = System.currentTimeMillis();
        System.out.format("*** %s %s\n", size, now - before);
        System.out.format("*** remove %s %s\n", nrm, rm_now - rm_before);
        before = now;
        inVec = coutv.toString();
        printTimestamp("loop_done");
      }
    } finally {
      engine.dispose();
    }
  }
}
